package com.nokia.analytics.avrotest.AvroPositioningETL;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.TimeZone;

import org.apache.avro.Schema;
import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.mapreduce.AvroKeyInputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.nokia.analytics.AvroTypes.Position;

public class AvroPositioningBinaryReader extends Configured implements Tool {

	public void printUsage() {
		System.out.println("Usage: hadoop jar path/to/this.jar "
				+ this.getClass() + " <input dir> <output dir>");

	}

	private static class AvroPositioningMapper extends
			Mapper<AvroKey<Position>, NullWritable, Text, NullWritable> {
		private Text prValue;

		@Override
		protected void setup(Context context) {
			prValue = new Text();
		}

		@Override
		protected void map(AvroKey<Position> key, NullWritable value,
				Context context) throws IOException, InterruptedException {

			Position pStatsDatum = key.datum();
			Date dateObj = new Date(Math.round(pStatsDatum
					.getCoordinatesTimestamp()));

			SimpleDateFormat parser = new SimpleDateFormat("yyyy/MM/dd/HH");
			parser.setTimeZone(TimeZone.getTimeZone("UTC"));
			String partitionDate = parser.format(dateObj);

			prValue.set("Partition Date: ["
					+ partitionDate
					+ "]"
					+ "\t"
					+ "Major Version: ["
					+ (pStatsDatum.getMajorVersion() != null ? pStatsDatum
							.getMajorVersion() : "")
					+ "]"
					+ "\t"
					+ "Minor Version: ["
					+ (pStatsDatum.getMinorVersion() != null ? pStatsDatum
							.getMinorVersion() : "")
					+ "]"
					+ "\t"
					+ "Time Stamp: ["
					+ (pStatsDatum.getCoordinatesTimestamp() != null ? pStatsDatum
							.getCoordinatesTimestamp() : "")
					+ "]"
					+ "\t"
					+ "Latitude: ["
					+ (pStatsDatum.getLatitude() != null ? pStatsDatum
							.getLatitude() : "")
					+ "]"
					+ "\t"
					+ "Longitude: ["
					+ (pStatsDatum.getLongitude() != null ? pStatsDatum
							.getLongitude() : "")
					+ "]"
					+ "\t"
					+ "Altitude: ["
					+ (pStatsDatum.getAltitude() != null ? pStatsDatum
							.getAltitude() : "")
					+ "]"
					+ "\t"
					+ "Haccuracy: ["
					+ (pStatsDatum.getHaccuracy() != null ? pStatsDatum
							.getHaccuracy() : "")
					+ "]"
					+ "\t"
					+ "Vaccuracy: ["
					+ (pStatsDatum.getVaccuracy() != null ? pStatsDatum
							.getVaccuracy() : "")
					+ "]"
					+ "\t"
					+ "Network Type:  ["
					+ (pStatsDatum.getNetworkType() != null ? pStatsDatum
							.getNetworkType() : "")
					+ "]"
					+ "\t"
					+ "MCC: ["
					+ (pStatsDatum.getServingMccSystemID() != null ? pStatsDatum
							.getServingMccSystemID() : "")
					+ "]"
					+ "\t"
					+ "MNC: ["
					+ (pStatsDatum.getServingMncNetworkID() != null ? pStatsDatum
							.getServingMncNetworkID() : "")
					+ "]"
					+ "\t"
					+ "Home MCC: ["
					+ (pStatsDatum.getHomeMccSystemID() != null ? pStatsDatum
							.getHomeMccSystemID() : "")
					+ "]"
					+ "\t"
					+ "Home MNC: ["
					+ (pStatsDatum.getHomeMncNetworkID() != null ? pStatsDatum
							.getHomeMncNetworkID() : "")
					+ "]"
					+ "\t"
					+ "Provided Measurements:  ["
					+ (pStatsDatum.getProvidedMeasurements() != null ? pStatsDatum
							.getProvidedMeasurements() : "")
					+ "]"
					+ "\t"
					+ "Used Measurements: ["
					+ (pStatsDatum.getUsedMeasurements() != null ? pStatsDatum
							.getUsedMeasurements() : "")
					+ "]"
					+ "\t"
					+ "Protocol Type: ["
					+ (pStatsDatum.getProtocolType() != null ? pStatsDatum
							.getProtocolType() : "")
					+ "]"
					+ "\t"
					+ "Protocol Version: ["
					+ (pStatsDatum.getProtocolVersion() != null ? pStatsDatum
							.getProtocolVersion() : "")
					+ "]"
					+ "\t"
					+ "Protocol Major Version: ["
					+ (pStatsDatum.getProtocolVersionMajor() != null ? pStatsDatum
							.getProtocolVersionMajor() : "")
					+ "]"
					+ "\t"
					+ "Protocol Minor Version: ["
					+ (pStatsDatum.getProtocolVersionMinor() != null ? pStatsDatum
							.getProtocolVersionMinor() : "")
					+ "]"
					+ "\t"
					+ "Manufacturer: ["
					+ (pStatsDatum.getManufacturer() != null ? pStatsDatum
							.getManufacturer() : "")
					+ "]"
					+ "\t"
					+ "Client Type: ["
					+ (pStatsDatum.getClientType() != null ? pStatsDatum
							.getClientType() : "")
					+ "]"
					+ "\t"
					+ "Status: ["
					+ (pStatsDatum.getStatus() != null ? pStatsDatum
							.getStatus() : "")
					+ "]"
					+ "\t"
					+ "Error Code: ["
					+ (pStatsDatum.getErrorCode() != null ? pStatsDatum
							.getErrorCode() : "")
					+ "]"
					+ "\t"
					+ "ModelId: ["
					+ (pStatsDatum.getModelId() != null ? pStatsDatum
							.getModelId() : "")
					+ "]"
					+ "\t"
					+ "Client Name: ["
					+ (pStatsDatum.getClientName() != null ? pStatsDatum
							.getClientName() : "")
					+ "]"
					+ "\t"
					+ "FQDN: ["
					+ (pStatsDatum.getFqdn() != null ? pStatsDatum.getFqdn()
							: "")
					+ "]"
					+ "\t"
					+ "SIM2 Network Type: ["
					+ (pStatsDatum.getSim2NetworkType() != null ? pStatsDatum
							.getSim2NetworkType() : "")
					+ "]"
					+ "\t"
					+ "SIM2 MCC: ["
					+ (pStatsDatum.getSim2ServingMccSystemID() != null ? pStatsDatum
							.getSim2ServingMccSystemID() : "")
					+ "]"
					+ "\t"
					+ "SIM2 MNC: ["
					+ (pStatsDatum.getSim2ServingMncNetworkID() != null ? pStatsDatum
							.getSim2ServingMncNetworkID() : "")
					+ "]"
					+ "\t"
					+ "SIM2 Home MCC: ["
					+ (pStatsDatum.getSim2HomeMccSystemID() != null ? pStatsDatum
							.getSim2HomeMccSystemID() : "")
					+ "]"
					+ "\t"
					+ "SIM2 Home MNC: ["
					+ (pStatsDatum.getSim2HomeMncNetworkID() != null ? pStatsDatum
							.getSim2HomeMncNetworkID() : "")
					+ "]"
					+ "\t"
					+ "SIM2 Provided Measurements: ["
					+ (pStatsDatum.getSim2ProvidedMeasurements() != null ? pStatsDatum
							.getSim2ProvidedMeasurements() : "")
					+ "]"
					+ "\t"
					+ "SIM2 Used Measurements: ["
					+ (pStatsDatum.getSim2UsedMeasurements() != null ? pStatsDatum
							.getSim2UsedMeasurements() : "") + "]");
			context.write(prValue, NullWritable.get());
		}
	}

	public int run(String[] args) throws IOException, InterruptedException,
			ClassNotFoundException {

		JobConf conf = new JobConf();
		String[] otherArgs = new GenericOptionsParser(conf, args)
				.getRemainingArgs();

		if (otherArgs.length != 2) {
			printUsage();
			System.exit(1);
		}

		Job job = new Job(conf, "Positioning binary reader");
		job.setJarByClass(AvroPositioningBinaryReader.class);

		job.setMapperClass(AvroPositioningMapper.class);
		job.setInputFormatClass(AvroKeyInputFormat.class);

		AvroJob.setInputKeySchema(job, new Position().getSchema());
		AvroJob.setInputValueSchema(job, Schema.create(Schema.Type.NULL));

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(NullWritable.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.setInputPaths(job, new Path(otherArgs[0]));
		Path outputPath = new Path(otherArgs[1]);
		FileOutputFormat.setOutputPath(job, outputPath);

		return job.waitForCompletion(true) ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {

		int res = ToolRunner.run(new Configuration(),
				new AvroPositioningBinaryReader(), args);
		System.exit(res);
	}
}